package com.shujia.flink.core

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

import java.time.Duration

object Demo5EventTIme {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //读取卡口过车数据
    val dataDS: DataStream[String] = env.socketTextStream("master", 8888)

    //整理数据取出道路编号和时间戳
    val kcDS: DataStream[(String, Long)] = dataDS.map(line => {
      val split: Array[String] = line.split(",")
      //道路编号
      val roadId: String = split(1)
      //时间戳
      val ts: Long = split(2).toLong
      (roadId, ts)
    })

    /**
     * 要使用事件时间需要告诉flink程序哪一个字段是事件时间
     * 时间字段必须是毫秒级别
     *
     */
    //默认水位线等于最新一条数据的时间戳，水位线只能增加不能减少
    //val assDS: DataStream[(String, Long)] = kcDS.assignAscendingTimestamps(kv => kv._2)

    val assDS: DataStream[(String, Long)] = kcDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //设置水位线的生成策略，前移5秒
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        //设置时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = {
            //时间字段
            element._2
          }
        })
    )

    /**
     * 统计每个道路的车流量 每隔5秒统计一次 统计最近5秒的车辆
     */
    val roadKvDS: DataStream[(String, Int)] = assDS.map(kv => (kv._1, 1))

    //按照道路分组
    val keyByDS: KeyedStream[(String, Int), String] = roadKvDS.keyBy(_._1)

    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyByDS
      //滑动的处理时间窗口
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))

    val countDS: DataStream[(String, Int)] = windowDS.sum(1)

    countDS.print()

    env.execute()
  }

}
